use bytes::{Buf, BufMut, Bytes, BytesMut};
use std::collections::HashMap;
use std::str;

use crate::error::{Result, VerticaError};

/// Message type constants
pub const AUTHENTICATION: u8 = b'R';
pub const BACKEND_KEY_DATA: u8 = b'K';
pub const BIND_COMPLETE: u8 = b'2';
pub const CLOSE_COMPLETE: u8 = b'3';
pub const COMMAND_COMPLETE: u8 = b'C';
pub const COPY_DATA: u8 = b'd';
pub const COPY_DONE: u8 = b'c';
pub const COPY_FAIL: u8 = b'f';
pub const COPY_IN_RESPONSE: u8 = b'G';
pub const COPY_OUT_RESPONSE: u8 = b'H';
pub const COPY_BOTH_RESPONSE: u8 = b'W';
pub const DATA_ROW: u8 = b'D';
pub const EMPTY_QUERY_RESPONSE: u8 = b'I';
pub const ERROR_RESPONSE: u8 = b'E';
pub const FUNCTION_CALL_RESPONSE: u8 = b'V';
pub const NO_DATA: u8 = b'n';
pub const NOTICE_RESPONSE: u8 = b'N';
pub const NOTIFICATION_RESPONSE: u8 = b'A';
pub const PARAMETER_DESCRIPTION: u8 = b't';
pub const PARAMETER_STATUS: u8 = b'S';
pub const PARSE_COMPLETE: u8 = b'1';
pub const PASSWORD_MESSAGE: u8 = b'p';
pub const PORTAL_SUSPENDED: u8 = b's';
pub const READY_FOR_QUERY: u8 = b'Z';
pub const ROW_DESCRIPTION: u8 = b'T';
pub const BIND: u8 = b'B';
pub const CLOSE: u8 = b'C';
pub const COPY_FAIL_CLIENT: u8 = b'f';
pub const DESCRIBE: u8 = b'D';
pub const EXECUTE: u8 = b'E';
pub const FLUSH: u8 = b'H';
pub const FUNCTION_CALL: u8 = b'F';
pub const PARSE: u8 = b'P';
pub const QUERY: u8 = b'Q';
pub const SYNC: u8 = b'S';
pub const TERMINATE: u8 = b'X';

/// Authentication request types
#[derive(Debug, Clone, PartialEq)]
pub enum AuthenticationType {
    Ok,
    KerberosV5,
    CleartextPassword,
    MD5Password { salt: [u8; 4] },
    SCM,
    GSS,
    SSPI,
    GSSContinue { data: Vec<u8> },
    SASL { mechanisms: Vec<String> },
    SASLContinue { data: Vec<u8> },
    SASLFinal { data: Vec<u8> },
}

impl AuthenticationType {
    pub fn from_code(code: i32, data: &[u8]) -> Result<Self> {
        match code {
            0 => Ok(AuthenticationType::Ok),
            2 => Ok(AuthenticationType::KerberosV5),
            3 => Ok(AuthenticationType::CleartextPassword),
            5 => {
                if data.len() >= 4 {
                    let mut salt = [0u8; 4];
                    salt.copy_from_slice(&data[..4]);
                    Ok(AuthenticationType::MD5Password { salt })
                } else {
                    Err(VerticaError::Protocol("Invalid MD5 salt".to_string()))
                }
            }
            6 => Ok(AuthenticationType::SCM),
            7 => Ok(AuthenticationType::GSS),
            9 => Ok(AuthenticationType::SSPI),
            8 => Ok(AuthenticationType::GSSContinue {
                data: data.to_vec(),
            }),
            10 => {
                let mechanisms = read_sasl_mechanisms(data)?;
                Ok(AuthenticationType::SASL { mechanisms })
            }
            11 => Ok(AuthenticationType::SASLContinue {
                data: data.to_vec(),
            }),
            12 => Ok(AuthenticationType::SASLFinal {
                data: data.to_vec(),
            }),
            _ => Err(VerticaError::Protocol(format!("Unknown authentication type: {}", code))),
        }
    }
}

/// Startup message
#[derive(Debug, Clone)]
pub struct StartupMessage {
    pub protocol_version: i32,
    pub parameters: HashMap<String, String>,
}

impl StartupMessage {
    pub fn new(parameters: HashMap<String, String>) -> Self {
        Self {
            protocol_version: 196608, // 3.0
            parameters,
        }
    }

    pub fn encode(&self) -> Result<Bytes> {
        let mut buf = BytesMut::new();
        
        buf.extend_from_slice(&self.protocol_version.to_be_bytes());
        
        for (key, value) in &self.parameters {
            buf.extend_from_slice(key.as_bytes());
            buf.put_u8(0);
            buf.extend_from_slice(value.as_bytes());
            buf.put_u8(0);
        }
        
        buf.put_u8(0);
        
        let total_len = buf.len() + 4;
        let mut result = BytesMut::with_capacity(total_len);
        result.extend_from_slice(&(total_len as i32).to_be_bytes());
        result.extend_from_slice(&buf);
        
        Ok(result.freeze())
    }
}

/// Query message
#[derive(Debug, Clone)]
pub struct QueryMessage {
    pub query: String,
}

impl QueryMessage {
    pub fn new(query: String) -> Self {
        Self { query }
    }

    pub fn encode(&self) -> Result<Bytes> {
        let mut buf = BytesMut::new();
        
        buf.extend_from_slice(self.query.as_bytes());
        buf.put_u8(0);
        
        let total_len = buf.len() + 4;
        let mut result = BytesMut::with_capacity(total_len + 1);
        result.put_u8(QUERY);
        result.extend_from_slice(&(total_len as i32).to_be_bytes());
        result.extend_from_slice(&buf);
        
        Ok(result.freeze())
    }
}

/// Parse message
#[derive(Debug, Clone)]
pub struct ParseMessage {
    pub name: String,
    pub query: String,
    pub param_types: Vec<u32>,
}

impl ParseMessage {
    pub fn new(name: String, query: String, param_types: Vec<u32>) -> Self {
        Self {
            name,
            query,
            param_types,
        }
    }

    pub fn encode(&self) -> Result<Bytes> {
        let mut buf = BytesMut::new();
        
        buf.extend_from_slice(self.name.as_bytes());
        buf.put_u8(0);
        buf.extend_from_slice(self.query.as_bytes());
        buf.put_u8(0);
        
        buf.extend_from_slice(&(self.param_types.len() as i16).to_be_bytes());
        for &param_type in &self.param_types {
            buf.extend_from_slice(&param_type.to_be_bytes());
        }
        
        let total_len = buf.len() + 4;
        let mut result = BytesMut::with_capacity(total_len + 1);
        result.put_u8(PARSE);
        result.extend_from_slice(&(total_len as i32).to_be_bytes());
        result.extend_from_slice(&buf);
        
        Ok(result.freeze())
    }
}

/// Bind message
#[derive(Debug, Clone)]
pub struct BindMessage {
    pub portal: String,
    pub statement: String,
    pub param_values: Vec<Option<Vec<u8>>>,
    pub result_formats: Vec<i16>,
}

impl BindMessage {
    pub fn new(
        portal: String,
        statement: String,
        param_values: Vec<Option<Vec<u8>>>,
        result_formats: Vec<i16>,
    ) -> Self {
        Self {
            portal,
            statement,
            param_values,
            result_formats,
        }
    }

    pub fn encode(&self) -> Result<Bytes> {
        let mut buf = BytesMut::new();
        
        buf.extend_from_slice(self.portal.as_bytes());
        buf.put_u8(0);
        buf.extend_from_slice(self.statement.as_bytes());
        buf.put_u8(0);
        
        buf.put_i16(0); // Parameter format codes (all text)
        
        buf.extend_from_slice(&(self.param_values.len() as i16).to_be_bytes());
        for param in &self.param_values {
            match param {
                Some(value) => {
                    buf.extend_from_slice(&(value.len() as i32).to_be_bytes());
                    buf.extend_from_slice(value);
                }
                None => buf.extend_from_slice(&(-1_i32).to_be_bytes()),
            }
        }
        
        buf.extend_from_slice(&(self.result_formats.len() as i16).to_be_bytes());
        for &format in &self.result_formats {
            buf.extend_from_slice(&format.to_be_bytes());
        }
        
        let total_len = buf.len() + 4;
        let mut result = BytesMut::with_capacity(total_len + 1);
        result.put_u8(BIND);
        result.extend_from_slice(&(total_len as i32).to_be_bytes());
        result.extend_from_slice(&buf);
        
        Ok(result.freeze())
    }
}

/// Execute message
#[derive(Debug, Clone)]
pub struct ExecuteMessage {
    pub portal: String,
    pub max_rows: i32,
}

impl ExecuteMessage {
    pub fn new(portal: String, max_rows: i32) -> Self {
        Self { portal, max_rows }
    }

    pub fn encode(&self) -> Result<Bytes> {
        let mut buf = BytesMut::new();
        
        buf.extend_from_slice(self.portal.as_bytes());
        buf.put_u8(0);
        buf.extend_from_slice(&self.max_rows.to_be_bytes());
        
        let total_len = buf.len() + 4;
        let mut result = BytesMut::with_capacity(total_len + 1);
        result.put_u8(EXECUTE);
        result.extend_from_slice(&(total_len as i32).to_be_bytes());
        result.extend_from_slice(&buf);
        
        Ok(result.freeze())
    }
}

/// Close message
#[derive(Debug, Clone)]
pub struct CloseMessage {
    pub kind: u8,
    pub name: String,
}

impl CloseMessage {
    pub fn new(kind: u8, name: String) -> Self {
        Self { kind, name }
    }

    pub fn encode(&self) -> Result<Bytes> {
        let mut buf = BytesMut::new();
        
        buf.put_u8(self.kind);
        buf.extend_from_slice(self.name.as_bytes());
        buf.put_u8(0);
        
        let total_len = buf.len() + 4;
        let mut result = BytesMut::with_capacity(total_len + 1);
        result.put_u8(CLOSE);
        result.extend_from_slice(&(total_len as i32).to_be_bytes());
        result.extend_from_slice(&buf);
        
        Ok(result.freeze())
    }
}

/// Sync message
#[derive(Debug, Clone)]
pub struct SyncMessage;

impl SyncMessage {
    pub fn encode(&self) -> Result<Bytes> {
        let mut result = BytesMut::with_capacity(5);
        result.put_u8(SYNC);
        result.extend_from_slice(&4_i32.to_be_bytes());
        
        Ok(result.freeze())
    }
}

/// Terminate message
#[derive(Debug, Clone)]
pub struct TerminateMessage;

impl TerminateMessage {
    pub fn encode(&self) -> Result<Bytes> {
        let mut result = BytesMut::with_capacity(5);
        result.put_u8(TERMINATE);
        result.extend_from_slice(&4_i32.to_be_bytes());
        
        Ok(result.freeze())
    }
}

/// Authentication response
#[derive(Debug, Clone)]
pub struct AuthenticationResponse {
    pub auth_type: AuthenticationType,
}

impl AuthenticationResponse {
    pub fn decode(mut buf: Bytes) -> Result<Self> {
        let auth_code = buf.get_i32();
        let data = buf.split_off(4);
        
        let auth_type = AuthenticationType::from_code(auth_code, &data)?;
        
        Ok(Self { auth_type })
    }
}

/// Backend key data
#[derive(Debug, Clone)]
pub struct BackendKeyData {
    pub process_id: i32,
    pub secret_key: i32,
}

impl BackendKeyData {
    pub fn decode(mut buf: Bytes) -> Result<Self> {
        let process_id = buf.get_i32();
        let secret_key = buf.get_i32();
        
        Ok(Self {
            process_id,
            secret_key,
        })
    }
}

/// Parameter status
#[derive(Debug, Clone)]
pub struct ParameterStatus {
    pub name: String,
    pub value: String,
}

impl ParameterStatus {
    pub fn decode(mut buf: Bytes) -> Result<Self> {
        let name = read_c_string(&mut buf)?;
        let value = read_c_string(&mut buf)?;
        
        Ok(Self { name, value })
    }
}

/// Row description
#[derive(Debug, Clone)]
pub struct RowDescription {
    pub fields: Vec<FieldDescription>,
}

#[derive(Debug, Clone)]
pub struct FieldDescription {
    pub name: String,
    pub table_oid: u32,
    pub column_attr: i16,
    pub type_oid: u32,
    pub type_size: i16,
    pub type_modifier: i32,
    pub format_code: i16,
}

impl RowDescription {
    pub fn decode(mut buf: Bytes) -> Result<Self> {
        let field_count = buf.get_i16() as usize;
        let mut fields = Vec::with_capacity(field_count);
        
        for _ in 0..field_count {
            let name = read_c_string(&mut buf)?;
            let table_oid = buf.get_u32();
            let column_attr = buf.get_i16();
            let type_oid = buf.get_u32();
            let type_size = buf.get_i16();
            let type_modifier = buf.get_i32();
            let format_code = buf.get_i16();
            
            fields.push(FieldDescription {
                name,
                table_oid,
                column_attr,
                type_oid,
                type_size,
                type_modifier,
                format_code,
            });
        }
        
        Ok(Self { fields })
    }
}

/// Parameter description
#[derive(Debug, Clone)]
pub struct ParameterDescription {
    pub type_oids: Vec<u32>,
}

impl ParameterDescription {
    pub fn decode(mut buf: Bytes) -> Result<Self> {
        let param_count = buf.get_i16() as usize;
        let mut type_oids = Vec::with_capacity(param_count);
        
        for _ in 0..param_count {
            type_oids.push(buf.get_u32());
        }
        
        Ok(Self { type_oids })
    }
}

/// Data row
#[derive(Debug, Clone)]
pub struct DataRow {
    pub values: Vec<Option<Bytes>>,
}

impl DataRow {
    pub fn decode(mut buf: Bytes) -> Result<Self> {
        let column_count = buf.get_i16() as usize;
        let mut values = Vec::with_capacity(column_count);
        
        for _ in 0..column_count {
            let len = buf.get_i32();
            if len < 0 {
                values.push(None);
            } else {
                let len = len as usize;
                let value = buf.split_to(len);
                values.push(Some(value));
            }
        }
        
        Ok(Self { values })
    }
}

/// Command complete
#[derive(Debug, Clone)]
pub struct CommandComplete {
    pub tag: String,
}

impl CommandComplete {
    pub fn decode(mut buf: Bytes) -> Result<Self> {
        let tag = read_c_string(&mut buf)?;
        Ok(Self { tag })
    }
}

/// Error response
#[derive(Debug, Clone)]
pub struct ErrorResponse {
    pub fields: HashMap<u8, String>,
}

impl ErrorResponse {
    pub fn decode(mut buf: Bytes) -> Result<Self> {
        let mut fields = HashMap::new();
        
        while buf.has_remaining() {
            let field_type = buf.get_u8();
            if field_type == 0 {
                break;
            }
            
            let value = read_c_string(&mut buf)?;
            fields.insert(field_type, value);
        }
        
        Ok(Self { fields })
    }

    pub fn message(&self) -> Option<&str> {
        self.fields.get(&b'M').map(|s| s.as_str())
    }
}

/// Ready for query
#[derive(Debug, Clone, PartialEq)]
pub enum TransactionStatus {
    Idle,
    Transaction,
    Error,
}

#[derive(Debug, Clone)]
pub struct ReadyForQuery {
    pub status: TransactionStatus,
}

impl ReadyForQuery {
    pub fn decode(mut buf: Bytes) -> Result<Self> {
        let status_byte = buf.get_u8();
        let status = match status_byte {
            b'I' => TransactionStatus::Idle,
            b'T' => TransactionStatus::Transaction,
            b'E' => TransactionStatus::Error,
            _ => return Err(VerticaError::Protocol(format!("Invalid transaction status: {}", status_byte))),
        };
        
        Ok(Self { status })
    }
}

/// Parse complete
#[derive(Debug, Clone)]
pub struct ParseComplete;

impl ParseComplete {
    pub fn decode(_buf: Bytes) -> Result<Self> {
        Ok(Self)
    }
}

/// Bind complete
#[derive(Debug, Clone)]
pub struct BindComplete;

impl BindComplete {
    pub fn decode(_buf: Bytes) -> Result<Self> {
        Ok(Self)
    }
}

/// Close complete
#[derive(Debug, Clone)]
pub struct CloseComplete;

impl CloseComplete {
    pub fn decode(_buf: Bytes) -> Result<Self> {
        Ok(Self)
    }
}

/// No data
#[derive(Debug, Clone)]
pub struct NoData;

impl NoData {
    pub fn decode(_buf: Bytes) -> Result<Self> {
        Ok(Self)
    }
}

/// Empty query response
#[derive(Debug, Clone)]
pub struct EmptyQueryResponse;

impl EmptyQueryResponse {
    pub fn decode(_buf: Bytes) -> Result<Self> {
        Ok(Self)
    }
}

fn read_sasl_mechanisms(data: &[u8]) -> Result<Vec<String>> {
    let s = String::from_utf8(data.to_vec())
        .map_err(|_| VerticaError::Connection("Invalid UTF-8 string".to_string()))?;
    Ok(s.split('\0').filter(|s| !s.is_empty()).map(|s| s.to_string()).collect())
}

fn read_c_string(buf: &mut Bytes) -> Result<String> {
    let mut bytes = Vec::new();
    
    while buf.has_remaining() {
        let byte = buf.get_u8();
        if byte == 0 {
            break;
        }
        bytes.push(byte);
    }
    
    String::from_utf8(bytes).map_err(|_| VerticaError::Connection("Invalid UTF-8 string".to_string()))
}

#[cfg(test)]
mod tests {
    use super::*;
    use bytes::BytesMut;

    #[test]
    fn test_startup_message() {
        let mut params = HashMap::new();
        params.insert("user".to_string(), "test".to_string());
        params.insert("database".to_string(), "testdb".to_string());
        
        let msg = StartupMessage::new(params);
        let encoded = msg.encode().unwrap();
        assert!(encoded.len() > 0);
    }

    #[test]
    fn test_query_message() {
        let msg = QueryMessage::new("SELECT 1".to_string());
        let encoded = msg.encode().unwrap();
        assert!(encoded.len() > 0);
    }

    #[test]
    fn test_parse_message() {
        let msg = ParseMessage::new("stmt1".to_string(), "SELECT $1".to_string(), vec![23]);
        let encoded = msg.encode().unwrap();
        assert!(encoded.len() > 0);
    }

    #[test]
    fn test_read_c_string() {
        let mut buf = BytesMut::new();
        buf.extend_from_slice(b"hello\0world\0");
        
        let mut bytes = buf.freeze();
        let s = read_c_string(&mut bytes).unwrap();
        assert_eq!(s, "hello");
        
        let s = read_c_string(&mut bytes).unwrap();
        assert_eq!(s, "world");
    }
}